package com.example.demo.kafka;

import com.example.demo.entity.OrderUtil;
import com.example.demo.services.OrderService;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class ConsumerListen {
    private Gson gson=new GsonBuilder().create();

    @Autowired
    private OrderService service;

    @KafkaListener(topics = "miaosha",concurrency = "5")
    public void listen(ConsumerRecord<String,String> record) throws Exception {
        Optional<?> KafkaMessage=Optional.ofNullable(record.value());
        String message=(String)KafkaMessage.get();
        OrderUtil util=gson.fromJson(message, OrderUtil.class);
        service.consumerFromKafaka(util.getStock(),util.getUsername());
    }
}